package com.T032;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

import java.io.IOException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * continentName	0
 * continentEnglishName	1
 * countryName	2
 * countryEnglishName	3
 * provinceName	4
 * provinceEnglishName	5
 * province_zipCode	6
 * province_confirmedCount	7
 * province_suspectedCount	8
 * province_curedCount	9
 * province_deadCount	10
 * cityName	11
 * cityEnglishName	12
 * city_zipCode	13
 * city_confirmedCount	14
 * city_suspectedCount	15
 * city_curedCount	16
 * city_deadCount	17
 * updateTime	18
 * <p>
 * <p>
 * 国外每天治愈人数最多的国家
 */
public class ByDayMaxCuredCountJob {

    public static class ByDayConfirmedCountMapper extends Mapper<Object, Text, Text, Text> {
        private Text outKey = new Text();
        private Text outValue = new Text();

        @Override
        protected void map(Object key, Text value, Context context) throws IOException, InterruptedException {
            String line = value.toString();
            if (line.startsWith("continentName")) {
                return;
            }
            String[] fields = line.split(",");

            //国家英文名
            String countryEnglishName = fields[3];

            //过滤中国
            if ("China".equals(countryEnglishName)) {
                return;
            }

            String day = fields[18].split(" ", -1)[0];


            outKey.set(day);

            context.write(outKey, value);
        }
    }

    public static class ByDayConfirmedCountReducer extends Reducer<Text, Text, Text, Text> {
        private Text outValue = new Text();

        Map<String, List<Integer>> map = new HashMap<>();

        @Override
        protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {

            int maxProvince_curedCount = 0;

            String maxContinentEnglishName = null;
            for (Text value : values) {
                String line = value.toString();
                String[] fields = line.split(",", -1);

                int province_curedCount = Integer.parseInt(fields[9]);

                if (province_curedCount > maxProvince_curedCount) {
                    maxProvince_curedCount = province_curedCount;
                    maxContinentEnglishName = fields[1];
                }
            }

            outValue.set(maxContinentEnglishName);
            context.write(key, outValue);
        }
    }

    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        Configuration conf = new Configuration();
        String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
        if (otherArgs.length < 2) {
            System.err.println("输入参数个数不正确！");
            System.exit(2);
        }
        Job job = new Job(conf, "AvgApp");
        job.setJarByClass(ByDayMaxCuredCountJob.class);
        job.setMapperClass(ByDayConfirmedCountMapper.class);
        job.setReducerClass(ByDayConfirmedCountReducer.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);
        FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
        FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
        Path outPath = new Path(otherArgs[1]);
        FileSystem fs = outPath.getFileSystem(conf);
        if (fs.exists(outPath)) {
            fs.delete(outPath, true);
        }
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}
